package cn.doitedu.kafka.day01

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 使用scala编写Kafka的生产者
 *
 * Producer是将数据发送到哪个分区中呢？
 *  默认是轮询写入
 * 可不可以指定将数据发送到指定的分区中
 *
 */
object ProducerDemo2 {

  def main(args: Array[String]): Unit = {


    //指定生产者的参数
    val properties = new Properties()
    //broker的地址
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //topic名称
    val topic = "test"
    //创建Kafka生产者
    val producer = new KafkaProducer[String, String](properties)

    //将数据封装到ProducerRecord中
    for(i <- 6001 to 8000) {

      //生成写入Kafka中的数据，用固定的字符串拼接上i
      //如果创建ProducerRecord，没有指定分区编号，默认的写入Topic分区的策略为轮询
      //但是不是写一条就切换，而是达到指定的大小才切换
      //目的是为了将数据均匀的写入到多个分区中，实现读写的负载均衡
      val record = new ProducerRecord[String, String](topic, "my data : "  + i)
      //发送数据
      producer.send(record)

    }

    producer.close()





  }

}
